package com.sleep.ws.server.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.sleep.ws.server.ChannelGroupPlus;
import com.sleep.ws.server.constant.WsServerConstant;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 * BaseWebSocketHandler
 * <p>
 * 消息前置处理
 */
public class BaseWebSocketHandler<I extends WebSocketFrame> extends SimpleChannelInboundHandler<WebSocketFrame> {

    private static final Logger log = LoggerFactory.getLogger(BaseWebSocketHandler.class);

    /**
     * 记录topic和订阅者的关系映射
     */
    private static final Map<String, ChannelGroupPlus> TOPIC_AND_CHANNEL_GROUP_MAPPING = new HashMap<>(100);

    public static Map<String, ChannelGroupPlus> getTopicAndChannelGroupMapping() {
        return TOPIC_AND_CHANNEL_GROUP_MAPPING;
    }

    public static void addTopic(String topic) {
        Arrays.stream(topic.split(",")).forEach(t -> {
            t = t.trim();
            if (StringUtils.hasLength(t)) {
                TOPIC_AND_CHANNEL_GROUP_MAPPING.putIfAbsent(t, new ChannelGroupPlus(new DefaultChannelGroup(GlobalEventExecutor.INSTANCE)));
            }
        });
    }

    /**
     * 消息执行的首要通道,起到转发相关的作用
     *
     * @param ctx 上下文环境
     * @param msg 消息内容
     * @throws Exception 子类异常
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame msg) throws Exception {
        if (msg instanceof PingWebSocketFrame) {
            ctx.writeAndFlush(new PongWebSocketFrame());
            return;
        }
        if (!(msg instanceof TextWebSocketFrame)) {
            ctx.channel().writeAndFlush(WebSocketCloseStatus.INVALID_MESSAGE_TYPE).addListener(ChannelFutureListener.CLOSE);
            return;
        }
        String text = ((TextWebSocketFrame) msg).text();
        JSONObject jsonObject;
        try {
            jsonObject = JSON.parseObject(text);
        } catch (JSONException e) {
            if ("ping".equals(text) || "heartCheck".equals(text)) {
                ctx.writeAndFlush(new TextWebSocketFrame("pong!"));
            }
            return;
        }
        if (jsonObject.containsKey(WsServerConstant.FLAG) && jsonObject.containsKey(WsServerConstant.TOPIC)) {
            //如果不存在topic映射，创建topic映射
            String topic = jsonObject.getString(WsServerConstant.TOPIC);
            addTopic(topic);

            ArrayList<String> topics = new ArrayList<>(50);
            if (topic.contains(",")) {
                topics.addAll(Arrays.asList(topic.split(",")));
            } else {
                topics.add(topic);
            }

            String flag = jsonObject.getString(WsServerConstant.FLAG);
            for (String tempTopic : topics) {
                switch (flag) {
                    //订阅消息处理
                    case WsServerConstant.CLIENT:
                        TOPIC_AND_CHANNEL_GROUP_MAPPING.computeIfPresent(tempTopic, (k, v) -> {
                            ChannelGroup topicChannels = v.getChannels();
                            Channel channel = ctx.channel();
                            InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
                            topicChannels.add(channel);
                            log.info("topic:{}添加消费者,地址{}", k, inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort());
                            return v;
                        });
                        break;
                    //push消息处理
                    case WsServerConstant.PRODUCT:
                        TOPIC_AND_CHANNEL_GROUP_MAPPING.computeIfPresent(tempTopic, (k, v) -> {
                            Channel channel = ctx.channel();
                            v.addProduct(channel);
                            if (!v.getChannels().isEmpty() && jsonObject.containsKey("data")) {
                                log.info("准备群发消息,topic:{}", topic);
                                TextWebSocketFrame pushMsg = new TextWebSocketFrame(jsonObject.get("data").toString());
                                onMessage(channel, TOPIC_AND_CHANNEL_GROUP_MAPPING.get(topic).getChannels(), pushMsg);
                            }
                            v.addCount();
                            v.upLastTimeStamp();
                            return v;
                        });
                        break;
                    default:
                }
            }
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state().equals(IdleState.WRITER_IDLE)) {
                ctx.writeAndFlush(new PongWebSocketFrame());
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.disconnect();
        ctx.fireChannelInactive();
        log.info("链接断开：{}-{}", ctx.channel().id(), ctx.channel().remoteAddress());
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        log.info("active ctx={}", ctx.name());
        log.info("链接创建：{}-{}", ctx.channel().id(), ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("异常发生", cause);
        ctx.close();
    }

    protected void onMessage(Channel currentChannel, ChannelGroup channels, TextWebSocketFrame msg) {
        // 如果是向客户端发送文本消息，则需要发送 TextWebSocketFrame 消息
        String txtMsg = msg.text();
        channels.forEach(ch -> {
            if (currentChannel != ch && !StringUtils.isEmpty(txtMsg)) {
                try {
                    ch.writeAndFlush(new TextWebSocketFrame(txtMsg));
                } catch (Exception e) {
                    log.error("通道错误，currentChannel,{},{}", currentChannel.localAddress().toString(), currentChannel.remoteAddress().toString());
                }
            }
        });

        log.debug("===发送完毕===");
    }
}
